﻿using FySystem.Logs;
using System;
using System.Buffers;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Text;
using System.Threading;
using System.Threading.Tasks;

namespace FySystem.IOTFlow
{
    public class WorkFlow<T>:IDisposable
    {
        private ManualResetEventSlim _allDone = new ManualResetEventSlim(false);
        private ConcurrentQueue<T> _queue = new ConcurrentQueue<T>();
        private Task _task = null;
        private Action<T[]> _exe = null;
        private bool _isRunning = true;

        /// <summary>
        /// 单数据推送
        /// </summary>
        public Action<T> OnDataPushedSingle { get; private set; }

        /// <summary>
        /// 多数据推送
        /// </summary>
        public Action<T[]> OnDataPushedArray { get; private set; }

        /// <summary>
        /// 多次取出数据时，一次最多取多少条
        /// </summary>
        public ushort MaxDequeueCount { get; set; } = 1000;

        public WorkFlow(Action<T> onDataPush)
        {
            OnDataPushedSingle = onDataPush;
            Init();
        }

        public WorkFlow(Action<T[]> onDataPush)
        {
            OnDataPushedArray = onDataPush;
            Init();
        }

        private void Init()
        {
            if (OnDataPushedSingle != null)
                _exe = HandleSingle;
            else
                _exe = HandleArray;

            _task = new Task(GetDataFromQueue);
            _task.Start();
        }

        /// <summary>
        /// 将数据放入队列
        /// </summary>
        /// <param name="data"></param>
        public void Enqueue(T data)
        {
            try
            {
                _queue.Enqueue(data);
                _allDone.Set();
            }
            catch (Exception ex)
            {
                FyLogContext.Write(ex.Message);
            }
        }

        /// <summary>
        /// 线程函数，从队列中获取数据
        /// </summary>
        private void GetDataFromQueue()
        {
            int num = 0;
            T[] items = null;

            while (_isRunning)
            {
                try
                {
                    _allDone.Reset();
                    num = _queue.Count;

                    if (num <= 0)
                    {
                        _allDone.Wait();
                        continue;
                    }

                    items = new T[num];
                    for(int i = 0; i < num; i++)
                    {
                        _queue.TryDequeue(out items[i]);
                    }

                    _exe(items);
                }
                catch (Exception ex)
                {
                    FyLogContextV2.WriteError("WorkFlow", ex.Message);
                }
            }
        }

        private void HandleArray(T[] items)
        {
            try
            {
                OnDataPushedArray.Invoke(items);
            }
            catch (Exception)
            {
                throw;
            }
        }

        private void HandleSingle(T[] items)
        {
            try
            {
                foreach (var item in items)
                    OnDataPushedSingle.Invoke(item);
            }
            catch (Exception)
            {
                throw;
            }
        }

        public void Dispose()
        {
            _isRunning = false;
        }
    }
}
